#include "Reactor.hpp"
#include "../base/Defind.hpp"
#include "Connection.hpp"
#include "Socket.hpp"
#include "../base/BaseLog.hpp"
#include "ConnPool.hpp"
#include "../base/ThreadPool.hpp"

#include <sys/epoll.h>
#include <unistd.h>
#include <cstdlib>
#include <cstring>

using namespace DBServer::Socket;


static int EpollAddOrMod(int epfd, int sock_fd, int flag, struct epoll_event* event) {
    event->events |= flag;
    if (epoll_ctl(epfd, EPOLL_CTL_ADD, sock_fd, event) == -1)
    {
        if (errno == EEXIST) // this socket fd already register in this instance
            epoll_ctl(epfd, EPOLL_CTL_MOD, sock_fd, event);
        else
            return -1;
    }
    return 0;
}

static int EpollDel(int epfd, int sock_fd, struct epoll_event* event) {
    return epoll_ctl(epfd, EPOLL_CTL_DEL, sock_fd, event);
}

static int EpollWait(int epfd, struct epoll_event* event) {
    int ret = epoll_wait(epfd, event, CONNCTION_MAX, 0);
    if (ret == -1)
    {
        if (errno == EINTR)
        {
            info("a siganl caugth\n");
            return 0;
        }
        else {
            error("epoll_wait fail!\n");
            return -1;
        }
    }
    return ret;
}

static int Read(int fd, char *buffer) {
    if (BUFFER_SIZE > SIZE_MAX)
    {
        error("what(): configure BUFFER_SIZE is too big!\n");
        exit(1);
    } else if (BUFFER_SIZE <= 0) {
        error("what(): configure BUFFER_SIZE <= 0!\n");
        exit(1);
    }
    // read the data fom the kernel tcp buffer
    int ret = read(fd, buffer, BUFFER_SIZE);
    if (ret == -1)
    {
        if (errno == EINTR)
        {
            // if it interrupted by a singal
            // read again
            ret = read(fd, buffer + strlen(buffer), BUFFER_SIZE) + ret;
        }
        else
        {
            error("what(): read() fail!\n");
            exit(1);
        }        
    }
    return ret;
}



/*--------------------------------------------
 * 
 * Reactor Api
 * 
 * -------------------------------------------
*/

using ThreadPool = DBServer::Base::ThreadPool;

Reactor::Reactor() 
    : status(REACTOR_FREE), 
      is_stop(0)
{
    reactor_init();
}

Reactor::~Reactor() {
    debug("");
}

// serilalize the data and push the data into the thread pool
int Reactor::run() {
    if (epoll_fd == -1)
    {
        if (reactor_init() == -1)
        {
            // the operation system has not enough file discriptor
            error("OS error - don't have enough file discriptor\n");
            exit(1);
        }
    }
    
    if (status == REACTOR_RUNING)
    {
        warn("  this Reactor is runing [%p]\n", (void*)this);
        return -1;
    }

    debug("run() begin!\n");

    // start the eventloop
    while (!is_stop)
    {
        struct epoll_event events[CONNCTION_MAX];

        int ret = EpollWait(epoll_fd, events);
        if (ret == 0 || ret == -1)
        {
            // if it not fd is ready or the system call is 
            // interrupted by a signal
            continue;
        }

        for (int i = 0; i < ret; i++) {
            int fd = events[i].data.fd;
            // find the connection by fd in the CONNECTED_POOL
            auto iter = map.find(fd);
            if (iter == map.end())
            {
                //maybe ... you know
                char format[] = "reactor(%d) the socket is not in CLIENT POOL! fd(%d)\n";
                warn(format, this->epoll_fd, fd);
                continue;
            }
            auto conn = iter->second;
            if (events[i].events & EPOLLRDHUP)
            {
err_or_close:
                // socket error and the socket is alread close
                delete_event(conn.get());     // delete the socket from epoll
                ConnPool::GetInstance()->back(std::move(conn));    // return back the socket to object pool
                map.erase(conn->get_socket_fd());   // push the connection from client pool
                socket_close(*conn);
                continue;
            }
            if (events[i].events & EPOLLIN)
            {
                debug("epoll in\n");

                int ret = 0;
                // multiple reactor moldou, do the tcp resolving job in loop
                std::vector<std::shared_ptr<Request>> set = conn->get_read_cb()(conn->get_socket_fd(), ret);
                if (ret == -1)
                    goto err_or_close;
                // commit the request into thread pool
                for (int i = 0; i < (int)set.size(); i++)
                { 
                    std::shared_ptr<Request> request = set[i];
                    // add the reference by lambda, 
                    ThreadPool::GetInstance()->commit([request, &conn]{
                        // user func call in the logic apect
                        std::string respond = conn->get_logic_cb()(request->s);
                        // call the send to write the respond  into the tcp write buffer
                        conn->get_buff()->write(respond);
                    });

                    // register a write event
                    add_eve_write(*conn);
                }
            }
            if (events[i].events & EPOLLOUT)
            {
                if (conn->get_write_cb()(conn->get_socket_fd()) == -1) 
                    goto err_or_close;

                // remote the write event
                remote_eve_write(*conn);
            }
        }
    }

    debug("Reactor::run() ending!\n");
    return 0;
}

int Reactor::stop() {
    is_stop = 1;
    {
        char format[] = "Reactor::run() ending!\n";
        debug(format);
    }
    return 0;
}

int Reactor::register_read(Connection* conn) {
    conn->set_epoll_fd(epoll_fd);
    conn->set_epoll_event();
    if (EpollAddOrMod(epoll_fd, conn->get_socket_fd(), EPOLLIN, conn->get_epoll_event()) == -1)
    {
        return -1;
    }
    ++monitor_fd;
    return 0;
}

int Reactor::register_write(Connection* conn) {
    conn->set_epoll_fd(epoll_fd);
    conn->set_epoll_event();
    if (EpollAddOrMod(epoll_fd, conn->get_socket_fd(), EPOLLOUT, conn->get_epoll_event()))
    {
        return -1;
    }
    return 0;
}

int Reactor::register_rdhup(Connection* conn) {
    conn->set_epoll_fd(epoll_fd);
    conn->set_epoll_event();
    if (EpollAddOrMod(epoll_fd, conn->get_socket_fd(), EPOLLRDHUP, conn->get_epoll_event()))
    {
        return -1;
    }
    return 0;
}

int Reactor::delete_event(Connection* conn) {
    if (EpollDel(epoll_fd, conn->get_socket_fd(), conn->get_epoll_event()))
    {
        return -1;
    }
    --monitor_fd;
    return 0;    
}

int Reactor::get_epoll_fd() {
    return epoll_fd;
}

int Reactor::reactor_init() {
    epoll_fd = epoll_create(100);
    if (epoll_fd == -1) {
        error("create a epoll fd fail !\n");
        exit(1);
    }
    return 0;
}

int Reactor::get_socket_num() {
    return monitor_fd;
}

void Reactor::add_conn(std::shared_ptr<ServerConn> conn) {
    monitor_fd++;
    map[conn->get_socket_fd()] = conn;
}

void Reactor::add_eve_write(Connection &conn) {
    EpollAddOrMod(epoll_fd, conn.get_socket_fd(), EPOLLOUT, conn.get_epoll_event());
}

void Reactor::remote_eve_write(Connection &conn) {
    struct epoll_event ev;
    ev.events = EPOLLOUT;
    EpollDel(epoll_fd, conn.get_socket_fd(), &ev);
}